// Licensed to the Apache Software Foundation (ASF) under one or more
// contributor license agreements.  See the NOTICE file distributed with
// this work for additional information regarding copyright ownership.
// The ASF licenses this file to You under the Apache License, Version 2.0
// (the "License"); you may not use this file except in compliance with
// the License.  You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use anyhow::{Context, Result};
use skywalking::reporter::CollectItem;
use std::{
    collections::VecDeque,
    fs::{self, File, OpenOptions},
    io::{self, Read, Write},
    path::{Path, PathBuf},
    time::{Duration, SystemTime, UNIX_EPOCH},
};
use tracing::{debug, info, warn};

// 文件名前缀
const FILE_PREFIX: &str = "skywalking-data-";
// 单个文件大小限制 (100MB)
const MAX_FILE_SIZE: u64 = 100 * 1024 * 1024;
// 默认TTL，超过此时间的数据将被删除 (1小时)
const DEFAULT_TTL_SECS: u64 = 3600;

// 存储统计信息
pub struct StorageStats {
    pub file_count: usize,
    pub total_size: u64,
    pub usage_percent: f64,
    pub oldest_file_time: Option<SystemTime>,
}

// 存储管理器，负责管理磁盘存储
pub struct StorageManager {
    storage: DiskStorage,
    // 恢复队列，每次从文件中加载一部分数据到内存中
    recovery_queue: VecDeque<CollectItem>,
    // 当前正在处理的文件
    current_recovery_file: Option<PathBuf>,
    // 存储文件列表，按时间排序
    file_list: Vec<PathBuf>,
    // 上次文件列表刷新时间
    last_refresh: SystemTime,
}

impl StorageManager {
    pub fn new(storage: DiskStorage) -> Self {
        Self {
            storage,
            recovery_queue: VecDeque::new(),
            current_recovery_file: None,
            file_list: Vec::new(),
            last_refresh: SystemTime::now(),
        }
    }

    // 存储一个数据项
    pub async fn store(&mut self, item: CollectItem) -> Result<()> {
        self.storage.store(item).await
    }

    // 批量恢复数据
    pub async fn recover_batch(&mut self, batch_size: usize) -> Result<Vec<CollectItem>> {
        let mut result = Vec::with_capacity(batch_size);
        
        // 如果恢复队列为空，尝试从文件中加载数据
        if self.recovery_queue.is_empty() {
            self.load_recovery_queue().await?;
        }
        
        // 从恢复队列中取出指定数量的数据
        for _ in 0..batch_size {
            if let Some(item) = self.recovery_queue.pop_front() {
                result.push(item);
            } else {
                // 队列已空，尝试加载更多数据
                if self.load_recovery_queue().await? == 0 {
                    // 没有更多数据可加载
                    break;
                }
                
                // 再次尝试获取数据
                if let Some(item) = self.recovery_queue.pop_front() {
                    result.push(item);
                } else {
                    break;
                }
            }
        }
        
        // 如果当前文件处理完毕，且恢复队列为空，删除已处理的文件
        if self.recovery_queue.is_empty() && self.current_recovery_file.is_some() {
            if let Some(path) = self.current_recovery_file.take() {
                if let Err(e) = fs::remove_file(&path) {
                    warn!(?e, ?path, "Failed to remove processed file");
                } else {
                    debug!(?path, "Removed processed file");
                }
            }
        }
        
        Ok(result)
    }
    
    // 加载数据到恢复队列
    async fn load_recovery_queue(&mut self) -> Result<usize> {
        // 至少每10秒刷新一次文件列表
        let now = SystemTime::now();
        if now.duration_since(self.last_refresh)? > Duration::from_secs(10) || self.file_list.is_empty() {
            self.refresh_file_list().await?;
            self.last_refresh = now;
        }
        
        // 如果没有文件可以恢复，直接返回
        if self.file_list.is_empty() {
            return Ok(0);
        }
        
        // 优先处理最早创建的文件
        let file_path = match self.current_recovery_file.take() {
            Some(path) if path.exists() => path,
            _ => {
                // 获取下一个文件
                match self.file_list.pop() {
                    Some(path) => path,
                    None => return Ok(0),
                }
            }
        };
        
        // 尝试从文件加载数据
        let items = self.storage.load_from_file(&file_path).await
            .with_context(|| format!("Failed to load data from file: {:?}", file_path))?;
        
        if items.is_empty() {
            // 文件为空或格式不正确，删除文件
            if let Err(e) = fs::remove_file(&file_path) {
                warn!(?e, ?file_path, "Failed to remove empty/invalid file");
            }
            return Ok(0);
        }
        
        // 将数据添加到恢复队列
        let count = items.len();
        self.recovery_queue.extend(items);
        self.current_recovery_file = Some(file_path);
        
        debug!(count, "Loaded items to recovery queue");
        Ok(count)
    }
    
    // 刷新文件列表
    async fn refresh_file_list(&mut self) -> Result<()> {
        let mut files = self.storage.list_files().await?;
        
        // 按文件创建时间排序（最旧的在最后）
        files.sort_by(|a, b| {
            extract_timestamp_from_filename(b)
                .cmp(&extract_timestamp_from_filename(a))
        });
        
        self.file_list = files;
        debug!(count = self.file_list.len(), "Refreshed storage file list");
        Ok(())
    }
    
    // 获取存储中的数据项数量估计值
    pub async fn stored_items_count(&self) -> usize {
        let count = self.recovery_queue.len();
        
        // 根据存储的大小估算数据项数量（假设每个项目平均1KB）
        // 这只是一个粗略估计，仅用于日志和监控
        let storage_size = match self.storage.total_size().await {
            Ok(size) => size,
            Err(_) => 0,
        };
        
        // 很粗略的估算，平均每项1KB
        let estimated_items = storage_size / 1024;
        
        count + estimated_items as usize
    }
    
    // 检查存储健康状态
    pub async fn check_health(&self) -> Result<StorageStats> {
        let stats = self.storage.stats().await?;
        
        // 如果使用率超过90%，发出警告
        if stats.usage_percent > 90.0 {
            warn!(
                usage_percent = stats.usage_percent,
                quota_bytes = self.storage.quota_bytes,
                "Storage usage is high"
            );
        }
        
        Ok(stats)
    }
}

// 从文件名中提取时间戳
fn extract_timestamp_from_filename(path: &Path) -> u64 {
    if let Some(file_name) = path.file_name() {
        if let Some(name_str) = file_name.to_str() {
            if name_str.starts_with(FILE_PREFIX) {
                if let Some(timestamp_str) = name_str.strip_prefix(FILE_PREFIX) {
                    if let Some(timestamp_end) = timestamp_str.find('-') {
                        if let Ok(timestamp) = timestamp_str[..timestamp_end].parse::<u64>() {
                            return timestamp;
                        }
                    }
                }
            }
        }
    }
    // 如果无法提取，返回0使其排在最后
    0
}

// 磁盘存储实现
pub struct DiskStorage {
    // 存储目录
    storage_path: PathBuf,
    // 存储配额（字节）
    pub quota_bytes: u64,
    // 数据TTL（秒）
    ttl_secs: u64,
}

impl DiskStorage {
    pub fn new(storage_path: PathBuf, quota_bytes: u64) -> Result<Self> {
        // 确保存储目录存在
        if !storage_path.exists() {
            fs::create_dir_all(&storage_path)?;
        }
        
        Ok(Self {
            storage_path,
            quota_bytes,
            ttl_secs: DEFAULT_TTL_SECS,
        })
    }
    
    // 设置数据TTL
    pub fn with_ttl(mut self, ttl_secs: u64) -> Self {
        self.ttl_secs = ttl_secs;
        self
    }
    
    // 存储单个数据项
    pub async fn store(&self, item: CollectItem) -> Result<()> {
        // 检查存储配额
        self.enforce_quota().await?;
        
        // 移除过期数据
        self.remove_expired_data().await?;
        
        // 创建文件名
        let timestamp = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs();
        let random_suffix = rand::random::<u32>();
        let file_name = format!("{}{}-{}.bin", FILE_PREFIX, timestamp, random_suffix);
        let file_path = self.storage_path.join(&file_name);
        
        // 检查当前文件大小，如果超过MAX_FILE_SIZE则创建新文件
        if let Ok(metadata) = fs::metadata(&file_path) {
            if metadata.len() >= MAX_FILE_SIZE {
                // 生成新的文件名
                let new_file_name = format!("{}-new", file_name);
                let new_file_path = self.storage_path.join(&new_file_name);
                
                // 将旧文件数据复制到新文件
                let mut old_file = File::open(&file_path)?;
                let mut new_file = File::create(&new_file_path)?;
                io::copy(&mut old_file, &mut new_file)?;
                
                // 删除旧文件
                fs::remove_file(&file_path)?;
                
                // 使用新文件路径
                return self.store_data_to_file(&new_file_path, &item).await;
            }
        }
        
        // 存储数据到文件
        self.store_data_to_file(&file_path, &item).await
    }
    
    // 提取存储逻辑到单独的方法
    async fn store_data_to_file(&self, file_path: &Path, item: &CollectItem) -> Result<()> {
        // 序列化数据
        let data = bincode::serde::encode_to_vec(item, bincode::config::standard())?;
        
        // 写入文件
        let mut file = OpenOptions::new()
            .write(true)
            .create(true)
            .truncate(true)
            .open(file_path)?;
        
        file.write_all(&data)?;
        file.sync_all()?;
        
        debug!(?file_path, size_bytes = data.len(), "Stored data to disk");
        Ok(())
    }
    
    // 从文件加载数据
    pub async fn load_from_file(&self, file_path: &Path) -> Result<Vec<CollectItem>> {
        let mut file = match File::open(file_path) {
            Ok(file) => file,
            Err(e) if e.kind() == io::ErrorKind::NotFound => {
                return Ok(Vec::new());
            }
            Err(e) => return Err(e.into()),
        };
        
        // 检查文件大小
        let metadata = file.metadata()?;
        if metadata.len() == 0 {
            return Ok(Vec::new());
        }
        
        // 读取文件内容
        let mut data = Vec::with_capacity(metadata.len() as usize);
        file.read_to_end(&mut data)?;
        
        // 反序列化数据
        match bincode::serde::decode_from_slice(&data, bincode::config::standard()) {
            Ok((item, _)) => Ok(vec![item]),
            Err(e) => {
                warn!(?e, ?file_path, "Failed to deserialize data");
                Ok(Vec::new())
            }
        }
    }
    
    // 列出所有存储文件
    pub async fn list_files(&self) -> Result<Vec<PathBuf>> {
        let mut result = Vec::new();
        
        for entry in fs::read_dir(&self.storage_path)? {
            let entry = entry?;
            let path = entry.path();
            
            // 检查是否为数据文件
            if is_data_file(&path) {
                result.push(path);
            }
        }
        
        Ok(result)
    }
    
    // 获取存储总大小
    pub async fn total_size(&self) -> Result<u64> {
        let mut total = 0;
        
        for entry in fs::read_dir(&self.storage_path)? {
            let entry = entry?;
            let metadata = entry.metadata()?;
            
            if metadata.is_file() && is_data_file(&entry.path()) {
                total += metadata.len();
            }
        }
        
        Ok(total)
    }
    
    // 获取存储统计信息
    pub async fn stats(&self) -> Result<StorageStats> {
        let mut file_count = 0;
        let mut total_size = 0;
        let mut oldest_time: Option<SystemTime> = None;
        
        for entry in fs::read_dir(&self.storage_path)? {
            let entry = entry?;
            let path = entry.path();
            
            if !is_data_file(&path) {
                continue;
            }
            
            let metadata = entry.metadata()?;
            if metadata.is_file() {
                file_count += 1;
                total_size += metadata.len();
                
                // 尝试从文件名提取时间戳
                let timestamp = extract_timestamp_from_filename(&path);
                if timestamp > 0 {
                    let file_time = UNIX_EPOCH + Duration::from_secs(timestamp);
                    
                    match oldest_time {
                        Some(current_oldest) if file_time < current_oldest => {
                            oldest_time = Some(file_time);
                        }
                        None => {
                            oldest_time = Some(file_time);
                        }
                        _ => {}
                    }
                }
            }
        }
        
        let usage_percent = if self.quota_bytes > 0 {
            (total_size as f64 / self.quota_bytes as f64) * 100.0
        } else {
            0.0
        };
        
        Ok(StorageStats {
            file_count,
            total_size,
            usage_percent,
            oldest_file_time: oldest_time,
        })
    }
    
    // 强制执行存储配额
    async fn enforce_quota(&self) -> Result<()> {
        // 获取当前存储大小
        let current_size = self.total_size().await?;
        
        // 如果超过90%配额，执行清理
        if current_size > (self.quota_bytes * 9 / 10) {
            debug!(
                current_size,
                quota_bytes = self.quota_bytes, 
                "Storage approaching quota, cleaning up old files"
            );
            
            // 获取所有文件并按时间排序（最旧的在前面）
            let mut files = self.list_files().await?;
            files.sort_by(|a, b| {
                extract_timestamp_from_filename(a)
                    .cmp(&extract_timestamp_from_filename(b))
            });
            
            // 计算需要释放的空间
            let target_size = self.quota_bytes * 7 / 10; // 释放到70%配额
            let mut size_to_free = if current_size > target_size {
                current_size - target_size
            } else {
                0
            };
            
            // 删除旧文件直到释放足够空间
            let mut freed = 0;
            for file in files {
                if size_to_free == 0 {
                    break;
                }
                
                match fs::metadata(&file) {
                    Ok(metadata) => {
                        let file_size = metadata.len();
                        if let Err(e) = fs::remove_file(&file) {
                            warn!(?e, ?file, "Failed to remove file during quota enforcement");
                        } else {
                            freed += 1;
                            if file_size <= size_to_free {
                                size_to_free -= file_size;
                            } else {
                                size_to_free = 0;
                            }
                        }
                    }
                    Err(e) => {
                        warn!(?e, ?file, "Failed to get file metadata during quota enforcement");
                    }
                }
            }
            
            info!(files_removed = freed, "Cleaned up old files to enforce storage quota");
        }
        
        Ok(())
    }
    
    // 删除过期数据
    async fn remove_expired_data(&self) -> Result<()> {
        if self.ttl_secs == 0 {
            // TTL禁用
            return Ok(());
        }
        
        let now = SystemTime::now();
        let cutoff = now - Duration::from_secs(self.ttl_secs);
        let cutoff_timestamp = cutoff.duration_since(UNIX_EPOCH)?.as_secs();
        
        let mut removed = 0;
        for file in self.list_files().await? {
            let timestamp = extract_timestamp_from_filename(&file);
            
            // 检查文件是否过期
            if timestamp > 0 && timestamp < cutoff_timestamp {
                if let Err(e) = fs::remove_file(&file) {
                    warn!(?e, ?file, "Failed to remove expired file");
                } else {
                    removed += 1;
                }
            }
        }
        
        if removed > 0 {
            debug!(removed, "Removed expired files");
        }
        
        Ok(())
    }
}

// 检查文件是否为数据文件
fn is_data_file(path: &Path) -> bool {
    if let Some(file_name) = path.file_name() {
        if let Some(file_name_str) = file_name.to_str() {
            return file_name_str.starts_with(FILE_PREFIX) && file_name_str.ends_with(".bin");
        }
    }
    false
} 